/**
 * Copyright (C) 2020 Baidu, Inc. All Rights Reserved.
 */
package com.baidu.mpks.backendweb.source.service;

import com.baidu.mpks.backendweb.datasource.service.HDFSService;
import com.baidu.mpks.backendweb.mongo.service.MongoService;
import com.baidu.mpks.backendweb.source.dao.DataSourceStatisticDayDao;
import com.baidu.mpks.backendweb.source.domain.DataSourceStatisticDayContentTypeVO;
import com.baidu.mpks.backendweb.source.domain.DataSourceStatisticDayDto;
import com.baidu.mpks.backendweb.source.domain.DataSourceStatisticDayInfo;
import com.baidu.mpks.backendweb.source.domain.DataSourceStatisticDayInputTypeVO;
import com.baidu.mpks.backendweb.source.domain.DataSourceStatisticDayTrendVO;
import com.baidu.mpks.datasource.constant.MongoKey;
import com.baidu.mpks.datasource.constant.SourceContentType;
import com.baidu.mpks.datasource.dao.SourceDao;
import com.baidu.mpks.datasource.domain.Source;
import com.baidu.mpks.util.MapUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.shiro.util.CollectionUtils;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Lazy;
import org.springframework.data.mongodb.core.MongoTemplate;
import org.springframework.data.mongodb.core.aggregation.Aggregation;
import org.springframework.data.mongodb.core.aggregation.AggregationResults;
import org.springframework.data.mongodb.core.query.Criteria;
import org.springframework.data.mongodb.core.query.Query;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.time.LocalDate;
import java.time.ZoneId;
import java.time.ZonedDateTime;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

/**
 * @author cuixuefei
 */
@Service
@Slf4j
public class SourceStatisticServiceImpl implements SourceStatisticService {

    @Autowired
    SourceDao sourceDao;

    @Value("${spring.data.mongodb.source.collection}")
    String sourceCollectionName;

    @Autowired
    @Lazy
    HDFSService hdfsService;

    @Resource
    DataSourceStatisticDayDao dataSourceStatisticDayDao;

    @Autowired
    MongoService mongoService;

    /**
     * 获取已生产接入数据，按照文件类型统计
     *
     * @return
     */
    @Override
    public List<DataSourceStatisticDayContentTypeVO> queryContentTypeData(List<String> contentTypes) {
        return dataSourceStatisticDayDao
                .findContentTypeDataByDate(contentTypes, new Date())
                .stream().map(item -> {
                    DataSourceStatisticDayContentTypeVO dataSourceStatisticDayContentTypeVO =
                            new DataSourceStatisticDayContentTypeVO();
                    BeanUtils.copyProperties(item, dataSourceStatisticDayContentTypeVO);
                    return dataSourceStatisticDayContentTypeVO;
                }).collect(Collectors.toList());
    }

    /**
     * 获取已生产数据分布情况，按照接入方式统计
     *
     * @return
     */
    @Override
    public List<DataSourceStatisticDayInputTypeVO> queryInputTypeData() {
        DataSourceStatisticDayDto dataSourceStatisticDayDto = new DataSourceStatisticDayDto();
        dataSourceStatisticDayDto.setDay(new java.sql.Date(getDateLong(new Date())));
        List<DataSourceStatisticDayInfo> dataSourceStatisticDayInfoList = dataSourceStatisticDayDao
                .findInputTypeData(dataSourceStatisticDayDto);
        List<DataSourceStatisticDayInputTypeVO> dataSourceStatisticDayInputTypeVOList = new ArrayList<>();
        for (DataSourceStatisticDayInfo dataSourceStatisticDayInfo : dataSourceStatisticDayInfoList) {
            DataSourceStatisticDayInputTypeVO dataSourceStatisticDayInputTypeVO =
                    new DataSourceStatisticDayInputTypeVO();
            BeanUtils.copyProperties(dataSourceStatisticDayInfo, dataSourceStatisticDayInputTypeVO);
            dataSourceStatisticDayInputTypeVOList.add(dataSourceStatisticDayInputTypeVO);
        }
        return dataSourceStatisticDayInputTypeVOList;
    }

    /**
     * 获取已生产数据趋势情况
     *
     * @param beginDate
     * @param endDate
     * @return
     */
    @Override
    public Map<String, List<DataSourceStatisticDayTrendVO>> queryDataTrend(long beginDate, long endDate) {
        DataSourceStatisticDayDto dataSourceStatisticDayDto = new DataSourceStatisticDayDto();
        dataSourceStatisticDayDto.setBeginDay(new java.sql.Date(beginDate));
        dataSourceStatisticDayDto.setEndDay(new java.sql.Date(endDate));
        List<DataSourceStatisticDayTrendVO> dataSourceStatisticDayTrendVOList = dataSourceStatisticDayDao
                .findTrendData(dataSourceStatisticDayDto);
        Map<String, List<DataSourceStatisticDayTrendVO>> mapSplit = splitDataToMap(dataSourceStatisticDayTrendVOList);
        Map<String, List<DataSourceStatisticDayTrendVO>> resultMap = coverData(mapSplit, beginDate, endDate);
        return resultMap;
    }

    /**
     * 将查询出来的数据按照contentType拆分，组装成图表的线
     *
     * @param listData
     * @return
     */
    private Map<String, List<DataSourceStatisticDayTrendVO>> splitDataToMap(
            List<DataSourceStatisticDayTrendVO> listData) {
        Map<String, List<DataSourceStatisticDayTrendVO>> resultMap = new HashMap<>();
        List<DataSourceStatisticDayTrendVO> resultDataList = null;
        for (DataSourceStatisticDayTrendVO dataSourceStatisticDayTrendVO : listData) {
            String contentType = dataSourceStatisticDayTrendVO.getContentType();
            if (resultMap.containsKey(contentType)) {
                resultDataList = resultMap.get(contentType);
            } else {
                resultDataList = new ArrayList<>();
                resultMap.put(contentType, resultDataList);
            }

            resultDataList.add(dataSourceStatisticDayTrendVO);
        }

        return resultMap;
    }

    private LocalDate date2LocalDate(Date date) {
        if (null == date) {
            return null;
        }
        return date.toInstant().atZone(ZoneId.systemDefault()).toLocalDate();
    }

    private Date localDate2Date(LocalDate localDate) {
        if (null == localDate) {
            return null;
        }
        ZonedDateTime zonedDateTime = localDate.atStartOfDay(ZoneId.systemDefault());
        return Date.from(zonedDateTime.toInstant());
    }

    /**
     * 将补齐的数据组装成map返回
     *
     * @param resultMap
     * @param beginDateL
     * @param endDateL
     * @return
     */
    private Map<String, List<DataSourceStatisticDayTrendVO>> coverData(
            Map<String, List<DataSourceStatisticDayTrendVO>> resultMap, long beginDateL, long endDateL) {
        LocalDate beginDate = date2LocalDate(new Date(beginDateL));
        LocalDate endDate = date2LocalDate(new Date(endDateL));
        Map<String, List<DataSourceStatisticDayTrendVO>> map = new HashMap<>();
        resultMap.forEach((k, v) -> {
            map.put(k, addDate(v, beginDate, endDate));
        });
        return map;
    }

    /**
     * 按照开始日期，结束日期补齐从数据库查出来缺少的日期数据
     *
     * @param list
     * @param beginDate
     * @param endDate
     * @return
     */
    private List<DataSourceStatisticDayTrendVO> addDate(List<DataSourceStatisticDayTrendVO> list, LocalDate beginDate,
                                                        LocalDate endDate) {
        List<DataSourceStatisticDayTrendVO> resultList = new ArrayList<>();
        while (beginDate.isBefore(endDate) || beginDate.isEqual(endDate)) {
            boolean flag = false;
            for (DataSourceStatisticDayTrendVO dataSourceStatisticDayTrendVO : list) {
                Date day = dataSourceStatisticDayTrendVO.getDay();
                LocalDate dayLocalDate = date2LocalDate(day);
                if (beginDate.isEqual(dayLocalDate)) {
                    flag = true;
                    resultList.add(dataSourceStatisticDayTrendVO);
                    break;
                }
            }

            if (!flag) {
                DataSourceStatisticDayTrendVO dataSourceStatisticDayTrendVO = new DataSourceStatisticDayTrendVO();
                dataSourceStatisticDayTrendVO.setTotal(0);
                dataSourceStatisticDayTrendVO.setDay(localDate2Date(beginDate));
                resultList.add(dataSourceStatisticDayTrendVO);
            }

            beginDate = beginDate.plusDays(1);
        }

        return resultList;
    }

    /**
     * 获取知识树节点已接入数据数量
     *
     * @param treeNodeList
     * @param contentType
     * @return
     */
    @Override
    public List<Map> queryNodeFilesTotal(List<Map> treeNodeList, String contentType) {
        for (Map treeNodeMap : treeNodeList) {
            Long total = 0L;
            Integer nodeId = (Integer) treeNodeMap.get("id");
            total = this.getNodeTotalByNodeId(nodeId, contentType);
            treeNodeMap.put("total", total);
        }

        return treeNodeList;
    }

    @Override
    public long getNodeTotalByNodeId(int nodeId, String contentType) {
        try {
            return sourceDao.sumSourceTotalByNodeIds(nodeId, contentType);
        } catch (Exception e) {
            return 0L;
        }
    }

    private Query getQueryObject(String keyCondition, List list) {
        Query query = null;
        if (list != null) {
            query = Query.query(Criteria.where(keyCondition).in(list));
        }

        return query;
    }

    private long getDateLong(Date date) {
        long dateL = 0;
        try {
            SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd");
            dateL = simpleDateFormat.parse(simpleDateFormat.format(date)).getTime();
        } catch (ParseException e) {
            log.error("" + e);
        }
        return dateL;
    }

    /**
     * 获取api列表接入数据数量
     *
     * @param sourceList
     * @param contentTypes
     * @param inputList
     * @return
     */
    @Override
    public List<Source> addSourceTotal(List<Source> sourceList, List<String> contentTypes, List<String> inputList) {
        List<Source> sourceTotalList = new ArrayList<>();
        if (!CollectionUtils.isEmpty(sourceList)) {
            List<Integer> sourceIdList = getSourceIds(sourceList);
            for (String contentType : contentTypes) {
                sourceTotalList = sourceDao.findByIdAndContentTypeAndInputTypes(sourceIdList,
                        contentType, inputList);
            }
        }
        return sourceTotalList;
    }

    private List<Integer> getSourceIds(List<Source> sourceList) {
        return sourceList.stream()
                .map(Source::getId)
                .collect(Collectors.toList());
    }

    @Override
    public void addSourceTotal() {
        SourceContentType[] contentTypes = SourceContentType.values();
        for (SourceContentType contentType : contentTypes) {
            if (!contentType.equals(SourceContentType.STRUCT)) {
                List<Source> sourceMongoTotalList = getMongoTotal(mongoService.getMongoTemplate(contentType));
                updateSourceTotal(contentType.getName(), sourceMongoTotalList);
            }
        }
    }

    //定时调用这块对source的total进行修改
    @Override
    public void updateSourceTotal(String contentType, List<Source> sources) {
        List<Source> sourceContentList = sourceDao.findByContentType(contentType);
        Map<Integer, Source> map = MapUtils.listToMap(Integer.class, Source.class, "getId", sources);
        Iterator<Source> iterator = sourceContentList.iterator();
        while (iterator.hasNext()) {
            Source source = iterator.next();
            Source s = map.get(source.getId());
            if (s != null) {
                source.setTotal(s.getTotal());
            } else {
                // 如果数据库中的sourceTotal已经是0了，那么移除，减少一次数据库操作
                if (source.getTotal() == 0) {
                    iterator.remove();
                } else {
                    source.setTotal(0);
                }
            }
        }
        sourceContentList.forEach(source -> sourceDao.updateSourceTotal(source.getId(),
                Long.valueOf(source.getTotal())));
    }

    private List<Source> getMongoTotal(MongoTemplate mongoTemplate) {
        Aggregation aggregation = Aggregation.newAggregation(
                Aggregation.match(Criteria.where(MongoKey.CRITERIA_MARK_DEL.getName()).is(0)),
                Aggregation.group(MongoKey.GROUP_FIELDS_SOURCE_ID.getName()).count()
                        .as(MongoKey.FIELDS_ALIAS_TOTAL.getName()));
        AggregationResults<Source> aggregationResults = mongoTemplate.aggregate(aggregation, sourceCollectionName,
                Source.class);
        return aggregationResults.getMappedResults();
    }

}
